/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2021/5/14.
//

#pragma once

#include <memory>
#include <vector>

#include "common/log/log.h"
#include "sql/parser/parse.h"
#include "sql/expr/tuple_cell.h"
#include "sql/expr/expression.h"
#include "storage/common/record.h"
#include <cmath>
#include <algorithm>

class Table;

class TupleCellSpec {
public:
    TupleCellSpec() = default;
    TupleCellSpec(Expression *expr) :
        expression_(expr) {
    }

    ~TupleCellSpec() {
        if (expression_) {
            delete expression_;
            expression_ = nullptr;
        }
    }

    void set_alias(const char *alias) {
        this->alias_ = alias;
    }
    const char *alias() const {
        return alias_;
    }

    Expression *expression() const {
        return expression_;
    }

private:
    const char *alias_ = nullptr;
    Expression *expression_ = nullptr;
};

class Tuple {
public:
    Tuple() = default;
    virtual ~Tuple() = default;

    virtual int cell_num() const = 0;
    virtual RC cell_at(int index, TupleCell &cell) const = 0;
    virtual RC find_cell(const Field &field, TupleCell &cell) const = 0;

    virtual RC cell_spec_at(int index, const TupleCellSpec *&spec) const = 0;
};

class RowTuple : public Tuple {
public:
    RowTuple() = default;
    virtual ~RowTuple() {
        for (TupleCellSpec *spec : speces_) {
            delete spec;
        }
        speces_.clear();
    }

    void set_record(Record *record) {
        this->record_ = record;
    }

    void set_schema(const Table *table, const std::vector<FieldMeta> *fields) {
        table_ = table;
        this->speces_.reserve(fields->size());
        for (const FieldMeta &field : *fields) {
            speces_.push_back(new TupleCellSpec(new FieldExpr(table, &field)));
        }
    }

    int cell_num() const override {
        return speces_.size();
    }

    RC cell_at(int index, TupleCell &cell) const override {
        if (index < 0 || index >= speces_.size()) {
            LOG_WARN("invalid argument. index=%d", index);
            return RC::INVALID_ARGUMENT;
        }

        const TupleCellSpec *spec = speces_[index];
        FieldExpr *field_expr = (FieldExpr *)spec->expression();
        const FieldMeta *field_meta = field_expr->field().meta();
        cell.set_type(field_meta->type());
        cell.set_data(this->record_->data() + field_meta->offset());
        cell.set_length(field_meta->len());
        return RC::SUCCESS;
    }

    RC find_cell(const Field &field, TupleCell &cell) const override {
        const char *table_name = field.table_name();
        if (0 != strcmp(table_name, table_->name())) {
            return RC::NOTFOUND;
        }

        const char *field_name = field.field_name();
        for (int i = 0; i < speces_.size(); ++i) {
            const FieldExpr *field_expr = (const FieldExpr *)speces_[i]->expression();
            const Field &field = field_expr->field();
            if (0 == strcmp(field_name, field.field_name())) {
                return cell_at(i, cell);
            }
        }
        return RC::NOTFOUND;
    }

    RC cell_spec_at(int index, const TupleCellSpec *&spec) const override {
        if (index < 0 || index >= speces_.size()) {
            LOG_WARN("invalid argument. index=%d", index);
            return RC::INVALID_ARGUMENT;
        }
        spec = speces_[index];
        return RC::SUCCESS;
    }

    Record &record() {
        return *record_;
    }

    const Record &record() const {
        return *record_;
    }

private:
    Record *record_ = nullptr;
    const Table *table_ = nullptr;
    std::vector<TupleCellSpec *> speces_;
};

/*
class CompositeTuple : public Tuple
{
public:
  int cell_num() const override;
  RC  cell_at(int index, TupleCell &cell) const = 0;
private:
  int cell_num_ = 0;
  std::vector<Tuple *> tuples_;
};
*/

class ProjectTuple : public Tuple {
public:
    ProjectTuple() = default;
    virtual ~ProjectTuple() {
        for (TupleCellSpec *spec : speces_) {
            delete spec;
        }
        speces_.clear();
    }

    void set_tuple(Tuple *tuple) {
        this->tuple_ = tuple;
    }

    void add_cell_spec(TupleCellSpec *spec) {
        speces_.push_back(spec);
    }
    int cell_num() const override {
        return speces_.size();
    }

    RC cell_at(int index, TupleCell &cell) const override {
        if (index < 0 || index >= speces_.size()) {
            return RC::GENERIC_ERROR;
        }
        if (tuple_ == nullptr) {
            return RC::GENERIC_ERROR;
        }

        const TupleCellSpec *spec = speces_[index];
        return spec->expression()->get_value(*tuple_, cell);
    }

    RC find_cell(const Field &field, TupleCell &cell) const override {
        return tuple_->find_cell(field, cell);
    }
    RC cell_spec_at(int index, const TupleCellSpec *&spec) const override {
        if (index < 0 || index >= speces_.size()) {
            return RC::NOTFOUND;
        }
        spec = speces_[index];
        return RC::SUCCESS;
    }

private:
    std::vector<TupleCellSpec *> speces_;
    Tuple *tuple_ = nullptr;
};

class TupleValue {
public:
    TupleValue() = default;
    virtual ~TupleValue() = default;

    virtual void to_string(std::ostream &os) const = 0;
    virtual int compare(const TupleValue &other) const = 0;
    virtual bool operator==(const TupleValue &value) const = 0;
    virtual bool operator>(const TupleValue &value) const = 0;
    virtual bool operator<(const TupleValue &value) const = 0;
};

class FloatValue : public TupleValue {
public:
    typedef float value_type;
    explicit FloatValue(float value) :
        value_(value) {
    }

    FloatValue(std::string value) :
        value_(atof(value.c_str())) {
    }

    void to_string(std::ostream &os) const override {
        int integer = int(value_);
        int frac = int(round((value_ + 1e-7) * 100)) % 100;
        os << frac * 0.01 + integer;
    }

    int compare(const TupleValue &other) const override {
        const FloatValue &float_other = (const FloatValue &)other;
        float result = value_ - float_other.value_;
        if (result > 0) { // 浮点数没有考虑精度问题
            return 1;
        }
        if (result < 0) {
            return -1;
        }
        return 0;
    }

    bool operator==(const TupleValue &other) const {
        const FloatValue &float_other = (const FloatValue &)other;
        return value_ == float_other.value_;
    }

    bool operator>(const TupleValue &other) const {
        const FloatValue &float_other = (const FloatValue &)other;
        return value_ > float_other.value_;
    }

    bool operator<(const TupleValue &other) const {
        const FloatValue &float_other = (const FloatValue &)other;
        return value_ < float_other.value_;
    }

public:
    void set_value(float value) {
        this->value_ = value;
    }

    float get_value() const {
        return value_;
    }

private:
    float value_;
};

class IntValue : public TupleValue {
public:
    typedef int value_type;
    explicit IntValue(int value) :
        value_(value) {
    }

    void to_string(std::ostream &os) const override {
        os << value_;
    }
    int compare(const TupleValue &other) const override {
        const IntValue &int_other = (const IntValue &)other;
        const FloatValue &float_other = (const FloatValue &)other;
        int compare_val;
        if (float_other.get_value() >= 0 && float_other.get_value() <= 0.000001)
            compare_val = int_other.value_;
        else { // ! 这种情况只会出现在avg里，因为avg默认存的都是FloatValue
            double right = float_other.get_value();
            return (double)value_ == right ? 0 : ((double)value_ > right ? 1 : -1);
        }
        return value_ == compare_val ? 0 : (value_ > compare_val ? 1 : -1);
    }
    bool operator==(const TupleValue &other) const {
        const IntValue &int_other = (const IntValue &)other;
        return value_ == int_other.value_;
    }

    bool operator>(const TupleValue &other) const {
        const IntValue &int_other = (const IntValue &)other;
        return value_ > int_other.value_;
    }

    bool operator<(const TupleValue &other) const {
        const IntValue &int_other = (const IntValue &)other;
        return value_ < int_other.value_;
    }

    void set_value(int value) {
        this->value_ = value;
    }

private:
    int value_;
};

class StringValue : public TupleValue {
public:
    typedef std::string value_type;
    explicit StringValue(const char *value, int len) :
        value_(value, len) {
    }
    explicit StringValue(const char *value) :
        value_(value) {
    }
    explicit StringValue(std::string value) :
        value_(value) {
    }
    explicit StringValue(float value, bool is_null = false) :
        value_(std::to_string(value)) {
    }

    void to_string(std::ostream &os) const override {
        os << value_;
    }

    int compare(const TupleValue &other) const override {
        const StringValue &string_other = (const StringValue &)other;
        return strcmp(value_.c_str(), string_other.value_.c_str());
    }

    bool operator==(const TupleValue &other) const {
        const StringValue &string_other = (const StringValue &)other;
        return strcmp(value_.c_str(), string_other.value_.c_str()) == 0;
    }

    bool operator>(const TupleValue &other) const {
        const StringValue &string_other = (const StringValue &)other;
        return strcmp(value_.c_str(), string_other.value_.c_str()) > 0;
    }

    bool operator<(const TupleValue &other) const {
        const StringValue &string_other = (const StringValue &)other;
        return strcmp(value_.c_str(), string_other.value_.c_str()) < 0;
    }

public:
    void set_value(const std::string &value) {
        this->value_ = value;
    }
    std::string get_value() {
        return this->value_;
    }

private:
    std::string value_;
};

class TupleDataRow {
public:
    void add(TupleValue *tuple_value) {
        values_.push_back(std::shared_ptr<TupleValue>(tuple_value));
    }

    void add(TupleCell &cell) {
        switch (cell.attr_type()) {
        case INTS: {
            add(new IntValue(*(int *)cell.data()));
        } break;
        case FLOATS: {
            add(new FloatValue(*(float *)cell.data()));
        } break;
        case CHARS: {
            std::string temp;
            for (int i = 0; i < cell.length(); i++) {
                if (cell.data()[i] == '\0') {
                    break;
                }
                temp += cell.data()[i];
            }
            add(new StringValue(temp));
        } break;
        }
    }

    void copy_and_delete(std::vector<int> order) {
        int del_size = values_.size();
        for (int i = 0; i < order.size(); i++) {
            add(values_[order[i]].get());
        }
        values_.erase(values_.begin(), values_.begin() + del_size);
    }

    void get(int index, std::shared_ptr<TupleValue> &tuple_value) {
        tuple_value = values_[index];
    }

    const std::shared_ptr<TupleValue> get(int index) {
        return values_[index];
    }

    std::vector<std::shared_ptr<TupleValue>> get_all_values() {
        return values_;
    }

    void merge_data_row(TupleDataRow &tuple_data_row) {
        for (auto &i : tuple_data_row.get_all_values()) {
            values_.push_back(i);
        }
    }

private:
    std::vector<std::shared_ptr<TupleValue>> values_;
};

class TupleSchema {
public:
    void set_fields(const std::vector<Field> &fields) {
        this->fields_.insert(this->fields_.end(), fields.begin(), fields.end());
    }

    const int get_length() {
        return fields_.size();
    }

    const void get(int index, Field &field) {
        field = fields_[index];
    }

    std::vector<Field> get_fields() {
        return fields_;
    }

    Field &get_field_by_index(int index) {
        return fields_[index];
    }

    void merge_schema(TupleSchema &schema) {
        for (auto &i : schema.get_fields()) {
            fields_.push_back(i);
        }
    }

    void set_schema(TupleSchema &schema) {
        fields_.clear();
        for (auto &i : schema.get_fields()) {
            fields_.push_back(i);
        }
    }

    void reverse_fields() {
        reverse(fields_.begin(), fields_.end());
    }

private:
    std::vector<Field> fields_;
};

class TupleSet {
public:
    void set_schema(TupleSchema &schema) {
        this->schema_ = schema;
    }

    TupleSchema &schema() {
        return schema_;
    }

    void add_value(TupleDataRow data) {
        this->data_.push_back(std::move(data));
    }

    int data_length() const {
        return data_.size();
    }

    void get_data_row(int index, TupleDataRow &data_row) {
        data_row = data_[index];
    }

    void to_string(std::ostream &os);

    void to_string(std::ostream &os, TupleSchema &schema);

    TupleSet merge_tuple_set(TupleSet &tuple_set, std::vector<Condition> &join_conditions);

    void project_tuple_set(TupleSet &tuple_set, TupleSchema &schema);

private:
    TupleSchema schema_;
    std::vector<TupleDataRow> data_;
};
